package com.peng.my_rocketmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSONObject;
import com.google.gson.JsonObject;
import com.peng.my_rocketmq.bean.User;






@Component
@RocketMQMessageListener
(topic = "peng-mq-spring-test", 
selectorType = SelectorType.TAG,
selectorExpression = "tagSend0",
consumerGroup = "${spring.application.name}-tagSend0")
public class Top0MessageExtConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {   

    @Override   

    // 实现消息的消费处理

    public void onMessage(MessageExt  messageExt) {
    	String key=messageExt.getKeys();
		String body=new String(messageExt.getBody());
    	
    	User message=(User) JSONObject.parse(body);
        System.out.printf("接受到------- :%s, body:%s key %s %n", message.getId(), message.getUsername(),key);   

    }   

    @Override

    // 设置从当前时间点开始消费消息  

    public void prepareStart(DefaultMQPushConsumer consumer) {

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);   
        consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));   

   }

}

